package org.ykhl.mq.commons;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.HashMap;
import java.util.Map;

/**
 * 用于在Kafka实现的MQ中传递的基础消息体实现。
 *
 * Created by zkf on 2015/11/25.
 */
public class BaseMessageKafkaImpl implements BaseMessage {
    private Map<String, String> map;

    public BaseMessageKafkaImpl(){
        map = new HashMap<>();
    }
    public BaseMessageKafkaImpl(String topic){
        this();
        setTopic(topic);
    }
    public BaseMessageKafkaImpl(String topic, String msgBody){
        this();
        setTopic(topic);
        setMessageBody(msgBody);
    }
    public void setMap(HashMap map){
        this.map = map;
    }
    @SuppressWarnings("unchecked")
    public BaseMessageKafkaImpl(ConsumerRecord<String, String> record){

        map = JSON.parseObject(record.value(), HashMap.class);
    }

    private String toJson(){
        return JSON.toJSONString(map);
    }

    public ProducerRecord<String, String> toProducerRecord(){
        return new ProducerRecord<>(getTopic(), toJson());
    }

    @Override
    public BaseMessageKafkaImpl newCopy() {
        BaseMessageKafkaImpl copy = new BaseMessageKafkaImpl();
        copy.map.putAll(map);
        return copy;
    }

    @Override
    public MessageType getMessageType() {
        String type = map.get("messageType").toString();
        return type == null ? null : MessageType.valueOf(type);
    }

    @Override
    public void setMessageType(MessageType messageType) {
        map.put("messageType", messageType.toString());
    }

    @Override
    public String getTopic() {
        return map.get("topic").toString();
    }

    @Override
    public void setTopic(String topic) {
        map.put("topic", topic);
    }

    @Override
    public String getHospitalId() {
        return map.get("hospitalId");
    }

    @Override
    public void setHospitalId(String hospitalId) {
        map.put("hospitalId", hospitalId);
    }

    @Override
    public String getHospitalName() {
        return map.get("hospitalName");
    }

    @Override
    public void setHospitalName(String hospitalName) {
        map.put("hospitalName", hospitalName);
    }

    @Override
    public String getDataType() {
        return map.get("dataType");
    }

    @Override
    public void setDataType(String dataType) {
        map.put("dataType", dataType);
    }

    @Override
    public String getDataVersion() {
        return map.get("version");
    }

    @Override
    public void setDataVersion(String version) {
        map.put("version", version);
    }

    @Override
    public String getAppServerId() {
        return map.get("appServerId");
    }

    @Override
    public void setAppServerId(String appServerId) {
        map.put("appServerId", appServerId);
    }

    @Override
    public String getAppServerName() {
        return map.get("appServerName");
    }

    @Override
    public void setAppServerName(String appServerName) {
        map.put("appServerName", appServerName);
    }

    @Override
    public String getAppId() {
        return map.get("appId");
    }

    @Override
    public void setAppId(String appId) {
        map.put("appId", appId);
    }

    @Override
    public String getMessageBody() {
        return map.get("messageBody");
    }

    @Override
    public void setMessageBody(String messageBody) {
        map.put("messageBody", messageBody);
        setMessageBodyMd5(Utils.genMd5(messageBody));
    }

    @Override
    public String getMessageBodyMd5() {
        return map.get("messageBodyMd5");
    }

    @Override
    public void setMessageBodyMd5(String messageBodyMd5) {
        map.put("messageBodyMd5", messageBodyMd5);
    }

    @Override
    public String getFlowId() {
        return map.get("flowId");
    }

    @Override
    public void setFlowId(String flowId) {
        map.put("flowId", flowId);
    }

    @Override
    public String getTimestamp() {
        return map.get("timestamp");
    }

    @Override
    public void setTimestamp(String timestamp) {
        map.put("timestamp", timestamp);
    }

    @Override
    public String getRequestId() {
        return map.get("requestId");
    }

    @Override
    public void setRequestId(String requestId) {
        map.put("requestId", requestId);
    }

    @Override
    public String getFid() {
        return map.get("fid");
    }

    @Override
    public void setFid(String fid) {
        map.put("fid", fid);
    }

    @Override
    public String getDeviceSign() {
        return map.get("deviceSign");
    }

    @Override
    public void setDeviceSign(String deviceSign) {
        map.put("deviceSign", deviceSign);
    }


    @Override
    public void setAppSecretKey(String appSecretKey) {
        map.put("appSecretKey", appSecretKey);
    }
    @Override
    public String getAppSecretKey() {

        return map.get("appSecretKey");
    }

    @Override
    public String getExtendSet() {
        return  map.get("extendSet");
    }
    @Override
    public void setExtendSet(String extendSet) {
        map.put("extendSet", extendSet);
    }

    @Override
    public String toString() {

        return "BaseMessageKafkaImpl{" +
                "messageType=" + getMessageType() +
                ", topic='" + getTopic() + '\'' +
                ", hospitalId='" + getHospitalId() + '\'' +
                ", hospitalName='" + getHospitalName() + '\'' +
                ", dataType='" + getDataType() + '\'' +
                ", dataVersion='" + getDataVersion() + '\'' +
                ", appServerId='" + getAppServerId() + '\'' +
                ", appServerName='" + getAppServerName() + '\'' +
                ", appId='" + getAppId() + '\'' +
                ", messageBodyMd5='" + getMessageBodyMd5() + '\'' +
                ", flowId='" + getFlowId() + '\'' +
                ", timestamp='" + getTimestamp() + '\'' +
                ", requestId='" + getRequestId() + '\'' +
                ", fid='" + getFid() + '\'' +
                ", deviceSign='" + getDeviceSign() + '\'' +
                ", appSecretKey='" + getAppSecretKey() + '\'' +
                ", extendSet='" + getExtendSet() + '\'' +
                ", messageBody='" + getMessageBody() + '\'' +
                '}';
    }
}
